package com.qiandw.common.cassandra

import java.util
import java.util.concurrent.ConcurrentHashMap

import com.datastax.driver.core.Session
import org.apache.commons.lang.StringUtils
import org.springframework.beans.factory.annotation.Autowired
import org.springframework.beans.factory.{DisposableBean, InitializingBean}
import org.springframework.stereotype.Component

import scala.beans.BeanProperty

@Component
class KeyspaceSessionFactoryImpl extends KeyspaceSessionFactory with InitializingBean with DisposableBean  {

  @Autowired
  var cfg :CassandraConfig = _

  @Autowired
  var cf: ClusterFactory = _

  @BeanProperty
  var defaultKeyspace: String = _

  @BeanProperty
  var keyspaces: util.List[String] = new util.ArrayList[String]

  private var defaultSession: Session = _

  @BeanProperty
  val sessionMap: util.Map[String, Session] = new ConcurrentHashMap[String, Session]

  override def getDefaultSession = {
    if (defaultSession == null || defaultSession.isClosed) try
      defaultSession = createSession(defaultKeyspace)
    catch {
      case e: Exception =>
    }
    this.defaultSession
  }

  override def getKeyspaceSession(keyspace: String) = {
    var session = sessionMap.get(keyspace)
    if (session == null || session.isClosed) try {
      session = createSession(keyspace)
      sessionMap.put(keyspace, session)
    } catch {
      case e: Exception =>
    }
    sessionMap.get(keyspace)
  }

  override def afterPropertiesSet() = {

    defaultKeyspace = cfg.defaultKeyspace

    defaultSession = createSession(defaultKeyspace)
    import scala.collection.JavaConversions._
    for (ks <- keyspaces) {
      if (!sessionMap.containsKey(ks)) if (ks == defaultKeyspace) sessionMap.put(ks, defaultSession)
      else {
        val s = createSession(ks)
        sessionMap.put(ks, s)
      }
    }

  }

  override def destroy(): Unit = {}

  private def createSession(keyspace: String) = {
    if (StringUtils.isEmpty(keyspace))
      cf.getCluster.connect.init
    else cf.getCluster.connect(keyspace).init
  }
}
